﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Text;

namespace Micro.RabbitMq
{
    /// <summary>
    /// RabbitMqHelper
    /// </summary>
    public sealed partial class RabbitMqHelper
    {
        #region 默认常量

        /// <summary>
        /// 默认连接字符串
        /// </summary>
        public const string DefaultConn = "amqp://sa:123456@127.0.0.1:5672/{0}";

        /// <summary>
        /// 默认虚拟主机
        /// </summary>
        public const string DefaultVir = "DefaultVir";

        /// <summary>
        /// 默认交换机
        /// </summary>
        public const string DefaultEx = "DefaultEx";

        /// <summary>
        /// 默认队列
        /// </summary>
        public const string DefaultQueue = "DefaultQueue";

        #endregion 默认常量

        /// <summary>
        /// 启动消费者数量
        /// </summary>
        private static int _consumerCount = 0;

        /// <summary>
        /// 连接池
        /// </summary>
        private static ConcurrentDictionary<string, IConnection> _connections = new ConcurrentDictionary<string, IConnection>();

        /// <summary>
        /// 信道池
        /// </summary>
        private static ConcurrentDictionary<string, IModel> _channels = new ConcurrentDictionary<string, IModel>();

        /// <summary>
        /// RabbitMq服务初始化
        /// </summary>
        public static void Initialize(string defaultConn = DefaultConn, string defaultVir = DefaultVir, string defaultEx = DefaultEx)
        {
            GetConn(defaultConn, defaultVir);
            GetChannel(defaultConn, defaultVir, defaultEx);
            AppDomain.CurrentDomain.ProcessExit += (sender, e) => Dispose();
        }

        /// <summary>
        /// 公用发送消息到队列
        /// </summary>
        /// <param name="message"></param>
        /// <param name="routingKey"></param>
        /// <param name="defaultEx"></param>
        /// <param name="channel"></param>
        /// <param name="defaultConn"></param>
        /// <param name="defaultVir"></param>
        /// <param name="defaultPriority"></param>
        /// <param name="defaultDelayMilliseconds"></param>
        /// <param name="isThrow"></param>
        /// <param name="isConsole"></param>
        /// <exception cref="Exception"></exception>
        public static void Publish(string message, string routingKey, string defaultEx = DefaultEx, IModel channel = default, string defaultConn = DefaultConn, string defaultVir = DefaultVir, byte defaultPriority = default, int defaultDelayMilliseconds = default, bool isThrow = false, bool isConsole = false)
        {
            try
            {
                channel = channel ?? GetChannel(defaultConn, defaultVir, defaultEx);
                var body = Encoding.UTF8.GetBytes(message);
                var properties = channel.CreateBasicProperties();
                properties.Persistent = true;
                properties.Priority = defaultPriority;
                if (defaultDelayMilliseconds > 0)
                    properties.Headers = new Dictionary<string, object>() { { "x-delay", defaultDelayMilliseconds } };
                channel.BasicPublish(exchange: defaultEx, routingKey: routingKey, mandatory: false, basicProperties: properties, body: body);
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                if (isThrow) throw new Exception(ex.Message);
            }
            if (isConsole) Console.WriteLine($"{DateTime.Now}:{defaultEx}:{routingKey}:发送一条消息！");
        }

        /// <summary>
        /// 获取连接
        /// </summary>
        /// <param name="defaultConn"></param>
        /// <param name="defaultVir"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public static IConnection GetConn(string defaultConn = DefaultConn, string defaultVir = DefaultVir)
        {
            if (string.IsNullOrEmpty(defaultConn)) throw new Exception("RabbitMq连接字符串不存在！");
            return _connections.GetOrAdd($"{defaultConn}_{defaultVir}", _ =>
            {
                var factory = new ConnectionFactory();
                factory.AutomaticRecoveryEnabled = true;
                factory.Uri = new Uri(string.Format(defaultConn, defaultVir));
                factory.RequestedHeartbeat = TimeSpan.Zero;
                return factory.CreateConnection();
            });
        }

        /// <summary>
        /// 获取信道
        /// </summary>
        /// <param name="defaultConn"></param>
        /// <param name="defaultVir"></param>
        /// <param name="defaultEx"></param>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public static IModel GetChannel(string defaultConn = DefaultConn, string defaultVir = DefaultVir, string defaultEx = DefaultEx)
        {
            return _channels.GetOrAdd($"{defaultConn}_{defaultVir}_{defaultEx}", _ => GetConn(defaultConn, defaultVir).CreateModel());
        }

        /// <summary>
        /// 初始化队列集合
        /// </summary>
        /// <param name="queueList"></param>
        /// <param name="defaultEx"></param>
        /// <param name="connection"></param>
        /// <param name="defaultConn"></param>
        /// <param name="defaultVir"></param>
        /// <param name="defaultExType"></param>
        /// <param name="defaultPriority"></param>
        /// <param name="isDelay"></param>
        /// <param name="isAsync"></param>
        public static async void InitQueues(IEnumerable<string> queueList = default, string defaultEx = DefaultEx, IConnection connection = default, string defaultConn = DefaultConn, string defaultVir = DefaultVir, string defaultExType = ExchangeType.Direct, byte defaultPriority = default, bool isDelay = default, bool isAsync = default)
        {
            var reTry = true;
            while (reTry)
            {
                try
                {
                    Initialize(defaultConn, defaultVir, defaultEx);
                    if (isAsync) await Task.Delay(1000);
                    if (queueList is null || !queueList.Any()) return;
                    connection = connection ?? GetConn(defaultConn, defaultVir);
                    using (var channel = connection.CreateModel())
                    {
                        //设置队列属性
                        var isArguments = defaultPriority > 0 ? true : false;
                        var queueArguments = new Dictionary<string, object>();
                        //设置队列优先级，取值范围在0~255之间。
                        if (defaultPriority > 0)
                            queueArguments.Add("x-max-priority", defaultPriority);
                        if (isDelay)
                            queueArguments.Add("x-delayed-type", defaultExType);
                        foreach (var queue in queueList)
                        {
                            if (isDelay)
                                //声名延时交换机
                                channel.ExchangeDeclare(exchange: defaultEx, type: "x-delayed-message", durable: true, false, queueArguments);
                            else
                                //声名交换机
                                channel.ExchangeDeclare(exchange: defaultEx, type: defaultExType, durable: true);

                            if (isArguments || isDelay)
                            {
                                //声名队列
                                channel.QueueDeclare(queue, true, default, default, queueArguments);
                            }
                            else
                            {
                                //声名队列
                                channel.QueueDeclare(queue, true, default, default);
                            }

                            //队列绑定交换机
                            channel.QueueBind(queue: queue, exchange: defaultEx, routingKey: queue);

                            //间隔一秒
                            Thread.Sleep(1000);
                        }
                    }
                    Console.WriteLine($"初始化创建队列{queueList.Count()}个成功");
                    reTry = false;
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"初始化创建队列异常：{ex.Message}");
                    // 等待一段时间后重试
                    Thread.Sleep(5000); // 5秒后重试连接
                }
            }
        }

        /// <summary>
        /// 初始化消费者
        /// </summary>
        /// <param name="action"></param>
        /// <param name="defaultQueue"></param>
        /// <param name="defaultConn"></param>
        /// <param name="defaultVir"></param>
        /// <param name="defaultPrefetchCount"></param>
        /// <param name="isConsole"></param>
        /// <param name="reQueue"></param>
        /// <param name="tryCount"></param>
        public static void InitConsumer(Action<string> action, string defaultQueue = DefaultQueue, string defaultConn = DefaultConn, string defaultVir = DefaultVir, ushort defaultPrefetchCount = 1, bool isConsole = true, bool reQueue = false, int tryCount = default)
        {
            var reTry = true;
            while (reTry)
            {
                IModel channel = default;
                try
                {
                    //创建通道
                    channel = GetConn(defaultConn, defaultVir).CreateModel();

                    #region 创建消费者监听

                    //公平分发、同一时间只处理一个消息。
                    channel.BasicQos(0, defaultPrefetchCount, false);

                    //创建消费者
                    var consumer = new EventingBasicConsumer(channel);

                    var logMsg = string.Empty;
                    consumer.Received += (model, ea) =>
                    {
                        var thisTryCount = tryCount;
                    gotoTryCount:
                        try
                        {
                            #region 初始化计时器

                            Stopwatch stopwatch = Stopwatch.StartNew();

                            #endregion 初始化计时器

                            #region 业务逻辑

                            var body = ea.Body.ToArray();
                            var message = Encoding.UTF8.GetString(body);
                            logMsg = message;
                            action.Invoke(message);

                            #endregion 业务逻辑

                            #region 计时结果

                            stopwatch.Stop();
                            if (isConsole)
                                Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff")}---消费消息（{defaultQueue}）:耗时：{stopwatch.ElapsedMilliseconds}毫秒");

                            #endregion 计时结果

                            //回复确认
                            channel.BasicAck(ea.DeliveryTag, false);
                        }
                        catch (Exception ex)
                        {
                            Thread.Sleep(1000);

                            //抛弃
                            if (!reQueue && --thisTryCount > 0) goto gotoTryCount;

                            //reQueue:true:重新放入队列,false:抛弃此条消息
                            channel.BasicNack(ea.DeliveryTag, false, reQueue);
                            Console.WriteLine($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff")}---【{(reQueue ? "重新放入队列" : "抛弃此条消息")}】【{defaultQueue}】【{logMsg}】一条消息：异常信息:{ex.Message}", true);
                        }
                    };
                    channel.BasicConsume(queue: defaultQueue,
                                            autoAck: false,
                                            consumer: consumer);

                    Console.WriteLine($"第{++_consumerCount}个消费者（{defaultQueue}）：启动成功", true);

                    //间隔一秒
                    Thread.Sleep(1000);
                    reTry = false;

                    #endregion 创建消费者监听
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"初始化创建消费者【{defaultQueue}】异常信息：{ex.Message}");

                    // 关闭连接和信道
                    channel?.Close();

                    // 等待一段时间后重试
                    Thread.Sleep(5000); // 5秒后重试连接
                }
            }
        }

        /// <summary>
        /// 应用程序关闭时释放【信道】【连接】
        /// </summary>
        private static void Dispose()
        {
            foreach (var channel in _channels)
            {
                try { channel.Value.Dispose(); } catch { }
            }

            foreach (var connection in _connections)
            {
                try { connection.Value.Dispose(); } catch { }
            }
        }
    }
}